package com.kang.product.tx.mq;

import com.alibaba.fastjson2.JSONObject;
import com.kang.product.tx.mapper.UserLoginLoggerMapper;
import com.kang.product.tx.mapper.UserLoginTransactionMapper;
import com.kang.product.tx.model.UserLoginLogger;
import com.kang.product.tx.model.UserLoginTransaction;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;

/**
 * 一个事物监听器对应一个事物流程,采取对应的rocketMQTemplate 发送事务消息
 */
@RocketMQTransactionListener(rocketMQTemplateBeanName = "myDiyMQTemplate")
public class PurchaseDressUpTransactionListener implements RocketMQLocalTransactionListener {

    @Autowired
    private UserLoginLoggerMapper userLoginLoggerMapper;

    @Autowired
    private UserLoginTransactionMapper userLoginTransactionMapper;

    /**
     * 执行本地事务
     * @param msg
     * @param arg
     * @return
     */
    @Override
    @Transactional
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 核心思路，登录日志(业务数据)　与本地事务状态表在同一个事务中
        // Object key = msg.getHeaders().get("key");
        UserLoginLogger loginLogger = JSONObject.parseObject
                (new String((byte[]) msg.getPayload()), UserLoginLogger.class);
        userLoginLoggerMapper.insert(loginLogger);
        UserLoginTransaction userLoginTransaction = new UserLoginTransaction();
        userLoginTransaction.setSerialNumber(loginLogger.getSerialNumber());
        userLoginTransaction.setCreateTime(new Date(System.currentTimeMillis()));
        userLoginTransactionMapper.insert(userLoginTransaction);
        //模拟异常,这里数据库事务由spring自动回滚
        int a = 1;
        // 把异常注悉
        /*if(a == 1) {
            throw  new RuntimeException();
        }*/
        return RocketMQLocalTransactionState.COMMIT;
    }

    /**
     * 检查本地事务状态
     *
     * @param msg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        try {
            UserLoginLogger loginLogger = JSONObject.parseObject(new String((byte[]) msg.getPayload()), UserLoginLogger.class);
            UserLoginTransaction queryModel = new UserLoginTransaction();
            queryModel.setSerialNumber(loginLogger.getSerialNumber());
            UserLoginTransaction transaction = userLoginTransactionMapper.selectOne(queryModel);
            if(transaction != null) { // 如果能查询到，则提交消息
                return RocketMQLocalTransactionState.COMMIT;
            }
        } catch (Exception e) {
            e.printStackTrace();
            return RocketMQLocalTransactionState.UNKNOWN;
            // return RocketMQLocalTransactionState.ROLLBACK;
        }
        // 如果事务状态表中未查询到，返回事务状态未知，RocketMQ默认在15次回查后将该消息回滚
        return RocketMQLocalTransactionState.UNKNOWN;
    }
}
